package com.atbeijing.D07;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * flink数据输出到redis
 *
 * 用虚拟机的时候请使用后台启动,关闭redis保护模式
 */
public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //redis主机地址
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("hadoop103").build();

        env
                .addSource(new SensorSource())
                .addSink(new RedisSink<SensorReading>(conf, new MyRedisSink()));

        env.execute();
    }

    public static class MyRedisSink implements RedisMapper<SensorReading>{

        @Override
        public RedisCommandDescription getCommandDescription() {
            //保存redis时 数据类型为hash,key为sensor
            return  new RedisCommandDescription(RedisCommand.HSET, "sensor");
        }

        @Override
        public String getKeyFromData(SensorReading data) {
            //保存redis时hash 的value 的key
            return data.id;
        }

        @Override
        public String getValueFromData(SensorReading data) {
            //保存redis时hset 的value 的value
            return data.temperature+"";
        }
    }

}
